/*-------------------------------------------------------------------------
 *
 * pgstatapprox.c
 *		  膨胀估算函数
 *
 * Copyright (c) 2014-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *		  contrib/pgstattuple/pgstatapprox.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/heapam.h"
#include "access/htup_details.h"
#include "access/multixact.h"
#include "access/relation.h"
#include "access/transam.h"
#include "access/visibilitymap.h"
#include "access/xact.h"
#include "catalog/namespace.h"
#include "catalog/pg_am_d.h"
#include "commands/vacuum.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "storage/bufmgr.h"
#include "storage/freespace.h"
#include "storage/lmgr.h"
#include "storage/procarray.h"
#include "utils/builtins.h"

PG_FUNCTION_INFO_V1(pgstattuple_approx);
PG_FUNCTION_INFO_V1(pgstattuple_approx_v1_5);

Datum		pgstattuple_approx_internal(Oid fc_relid, FunctionCallInfo fc_fcinfo);

typedef struct output_type
{
	uint64		table_len;
	uint64		scanned_percent;
	uint64		tuple_count;
	uint64		tuple_len;
	double		tuple_percent;
	uint64		dead_tuple_count;
	uint64		dead_tuple_len;
	double		dead_tuple_percent;
	uint64		free_space;
	double		free_percent;
} output_type;

#define NUM_OUTPUT_COLUMNS 10

/*
 * 此函数接受一个已经打开的关系并扫描其页面，
 * 跳过那些对应的可见性映射位被设置的页面。
 * 对于我们跳过的页面，我们从空闲空间映射中找到
 * 空闲空间并根据此进行元组长度的近似。对于其他页面，我们统计
 * 精确的死元组数量等。
 *
 * 此扫描松散地基于vacuumlazy.c:lazy_scan_heap()，但
 * 我们不尝试避免跳过单个页面。
 */
static void fc_statapprox_heap(Relation fc_rel, output_type *fc_stat)
{
	BlockNumber fc_scanned,
				fc_nblocks,
				fc_blkno;
	Buffer		fc_vmbuffer = InvalidBuffer;
	BufferAccessStrategy fc_bstrategy;
	TransactionId fc_OldestXmin;

	fc_OldestXmin = GetOldestNonRemovableTransactionId(fc_rel);
	fc_bstrategy = GetAccessStrategy(BAS_BULKREAD);

	fc_nblocks = RelationGetNumberOfBlocks(fc_rel);
	fc_scanned = 0;

	for (fc_blkno = 0; fc_blkno < fc_nblocks; fc_blkno++)
	{
		Buffer		fc_buf;
		Page		fc_page;
		OffsetNumber fc_offnum,
					fc_maxoff;
		Size		fc_freespace;

		CHECK_FOR_INTERRUPTS();

		/*
		 * 如果页面中只有可见元组，则我们可以从空闲空间映射中
		 * 找到空闲空间并继续。
		 */
		if (VM_ALL_VISIBLE(fc_rel, fc_blkno, &fc_vmbuffer))
		{
			fc_freespace = GetRecordedFreeSpace(fc_rel, fc_blkno);
			fc_stat->tuple_len += BLCKSZ - fc_freespace;
			fc_stat->free_space += fc_freespace;
			continue;
		}

		fc_buf = ReadBufferExtended(fc_rel, MAIN_FORKNUM, fc_blkno,
								 RBM_NORMAL, fc_bstrategy);

		LockBuffer(fc_buf, BUFFER_LOCK_SHARE);

		fc_page = BufferGetPage(fc_buf);

		/*
		 * 在新页面上调用PageGetHeapFreeSpace()是不安全的，因此我们
		 * 将它们视为空闲空间来处理。
		 */
		if (!PageIsNew(fc_page))
			fc_stat->free_space += PageGetHeapFreeSpace(fc_page);
		else
			fc_stat->free_space += BLCKSZ - SizeOfPageHeaderData;

		/* 即使页面是新创建的/空的，我们也可以将其计为已扫描 */
		fc_scanned++;

		if (PageIsNew(fc_page) || PageIsEmpty(fc_page))
		{
			UnlockReleaseBuffer(fc_buf);
			continue;
		}

		/*
		 * 查看页面上的每个元组并决定其是活动的还是死的，
		 * 然后进行计数并记录其大小。与lazy_scan_heap不同，我们可以
		 * 忽略问题和特殊情况。
		 */
		fc_maxoff = PageGetMaxOffsetNumber(fc_page);

		for (fc_offnum = FirstOffsetNumber;
			 fc_offnum <= fc_maxoff;
			 fc_offnum = OffsetNumberNext(fc_offnum))
		{
			ItemId		fc_itemid;
			HeapTupleData fc_tuple;

			fc_itemid = PageGetItemId(fc_page, fc_offnum);

			if (!ItemIdIsUsed(fc_itemid) || ItemIdIsRedirected(fc_itemid) ||
				ItemIdIsDead(fc_itemid))
			{
				continue;
			}

			Assert(ItemIdIsNormal(fc_itemid));

			ItemPointerSet(&(fc_tuple.t_self), fc_blkno, fc_offnum);

			fc_tuple.t_data = (HeapTupleHeader) PageGetItem(fc_page, fc_itemid);
			fc_tuple.t_len = ItemIdGetLength(fc_itemid);
			fc_tuple.t_tableOid = RelationGetRelid(fc_rel);

			/*
			 * 我们在计数INSERT_IN_PROGRESS元组时遵循VACUUM的做法，
			 * 将其视为“死”；而DELETE_IN_PROGRESS元组则视为“活”。我们不
			 * 费力区分由我们自己的事务插入/删除的元组。
			 */
			switch (HeapTupleSatisfiesVacuum(&fc_tuple, fc_OldestXmin, fc_buf))
			{
				case HEAPTUPLE_LIVE:
				case HEAPTUPLE_DELETE_IN_PROGRESS:
					fc_stat->tuple_len += fc_tuple.t_len;
					fc_stat->tuple_count++;
					break;
				case HEAPTUPLE_DEAD:
				case HEAPTUPLE_RECENTLY_DEAD:
				case HEAPTUPLE_INSERT_IN_PROGRESS:
					fc_stat->dead_tuple_len += fc_tuple.t_len;
					fc_stat->dead_tuple_count++;
					break;
				default:
					elog(ERROR, "unexpected HeapTupleSatisfiesVacuum result");
					break;
			}
		}

		UnlockReleaseBuffer(fc_buf);
	}

	fc_stat->table_len = (uint64) fc_nblocks * BLCKSZ;

	/*
	 * 对于我们未扫描的页面，我们不知道有多少元组，
	 * 因此以与VACUUM相同的方式将活动元组数推断到整个表中。
	 * （与VACUUM一样，我们并没有随机抽样，因此仅仅线性推断似乎不安全。）
	 * 所有可见页面中不应有死元组，因此不需要对此进行修正，我们
	 * 也已考虑这些页面中的空间。
	 */
	fc_stat->tuple_count = vac_estimate_reltuples(fc_rel, fc_nblocks, fc_scanned,
											   fc_stat->tuple_count);

	/* 这里不清楚我们是否会得到-1，但要小心。 */
	fc_stat->tuple_count = Max(fc_stat->tuple_count, 0);

	/*
	 * 如果关系有一个或多个页面，则计算百分比。
	 */
	if (fc_nblocks != 0)
	{
		fc_stat->scanned_percent = 100 * fc_scanned / fc_nblocks;
		fc_stat->tuple_percent = 100.0 * fc_stat->tuple_len / fc_stat->table_len;
		fc_stat->dead_tuple_percent = 100.0 * fc_stat->dead_tuple_len / fc_stat->table_len;
		fc_stat->free_percent = 100.0 * fc_stat->free_space / fc_stat->table_len;
	}

	if (BufferIsValid(fc_vmbuffer))
	{
		ReleaseBuffer(fc_vmbuffer);
		fc_vmbuffer = InvalidBuffer;
	}
}

/*
 * 返回给定relid的估计活动/死元组统计信息。
 *
 * 此处的superuser()检查必须保留，因为库可能会在没有
 * 升级扩展的情况下进行升级，这意味着在1.5之前的安装中
 * 这些函数可以被任何用户调用。
 */
Datum pgstattuple_approx(PG_FUNCTION_ARGS)
{
	Oid			fc_relid = PG_GETARG_OID(0);

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to use pgstattuple functions")));

	PG_RETURN_DATUM(pgstattuple_approx_internal(fc_relid, fcinfo));
}

/*
 * 自pgstattuple版本1.5以来，我们不再需要检查用户
 * 是否为超级用户，因为我们已对公共用户撤销了
 * SQL函数的执行权限。
 * 用户可以根据他们的策略授予访问权限。
 *
 * 与pgstattuple_approx（上述）基本相同。
 */
Datum pgstattuple_approx_v1_5(PG_FUNCTION_ARGS)
{
	Oid			fc_relid = PG_GETARG_OID(0);

	PG_RETURN_DATUM(pgstattuple_approx_internal(fc_relid, fcinfo));
}

Datum pgstattuple_approx_internal(Oid fc_relid, FunctionCallInfo fc_fcinfo)
{
	Relation	fc_rel;
	output_type fc_stat = {0};
	TupleDesc	fc_tupdesc;
	bool		fc_nulls[NUM_OUTPUT_COLUMNS];
	Datum		fc_values[NUM_OUTPUT_COLUMNS];
	HeapTuple	fc_ret;
	int			fc_i = 0;

	if (get_call_result_type(fc_fcinfo, NULL, &fc_tupdesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "return type must be a row type");

	if (fc_tupdesc->natts != NUM_OUTPUT_COLUMNS)
		elog(ERROR, "incorrect number of output arguments");

	fc_rel = relation_open(fc_relid, AccessShareLock);

	/*
	 * 拒绝尝试读取非本地临时关系；我们可能会获取错误的数据，因为
	 * 我们无法查看拥有会话的本地缓冲区。
	 */
	if (RELATION_IS_OTHER_TEMP(fc_rel))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("cannot access temporary tables of other sessions")));

	/*
	 * 我们仅支持具有可见性映射和空闲空间映射的关系类型。
	 */
	if (!(fc_rel->rd_rel->relkind == RELKIND_RELATION ||
		  fc_rel->rd_rel->relkind == RELKIND_MATVIEW ||
		  fc_rel->rd_rel->relkind == RELKIND_TOASTVALUE))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("relation \"%s\" is of wrong relation kind",
						RelationGetRelationName(fc_rel)),
				 errdetail_relkind_not_supported(fc_rel->rd_rel->relkind)));

	if (fc_rel->rd_rel->relam != HEAP_TABLE_AM_OID)
		ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
						errmsg("only heap AM is supported")));

	fc_statapprox_heap(fc_rel, &fc_stat);

	relation_close(fc_rel, AccessShareLock);

	memset(fc_nulls, 0, sizeof(fc_nulls));

	fc_values[fc_i++] = Int64GetDatum(fc_stat.table_len);
	fc_values[fc_i++] = Float8GetDatum(fc_stat.scanned_percent);
	fc_values[fc_i++] = Int64GetDatum(fc_stat.tuple_count);
	fc_values[fc_i++] = Int64GetDatum(fc_stat.tuple_len);
	fc_values[fc_i++] = Float8GetDatum(fc_stat.tuple_percent);
	fc_values[fc_i++] = Int64GetDatum(fc_stat.dead_tuple_count);
	fc_values[fc_i++] = Int64GetDatum(fc_stat.dead_tuple_len);
	fc_values[fc_i++] = Float8GetDatum(fc_stat.dead_tuple_percent);
	fc_values[fc_i++] = Int64GetDatum(fc_stat.free_space);
	fc_values[fc_i++] = Float8GetDatum(fc_stat.free_percent);

	fc_ret = heap_form_tuple(fc_tupdesc, fc_values, fc_nulls);
	return HeapTupleGetDatum(fc_ret);
}
